package com.gzhryc.common.kafka.consumer;

import java.time.Duration;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicBoolean;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import com.gzhryc.common.kafka.JFKafkaConsumer;
import com.gzhryc.common.kafka.consumer.interfaces.KafkaConsumerEventListener;
import com.gzhryc.common.kafka.models.ConsumerEvent;

/**
 * 多消费服务
 * @author zhanghao
 * @version 2020-9-23
 */
public class MultipleKafkaConsumer implements JFKafkaConsumer {

	private String topic;
	private String groupId;
	private Integer consumerCount;
	private KafkaConsumerEventListener listener;

	private Map<Integer, ConsumerThread> consumerList;
	private AtomicBoolean isRunning;

	public MultipleKafkaConsumer(String topic, Integer consumerCount, KafkaConsumerEventListener listener) {
		this.topic = topic;
		this.groupId = topic;
		this.consumerCount = consumerCount;
		this.listener = listener;
		this.consumerList = new ConcurrentHashMap<Integer, ConsumerThread>();
		this.isRunning = new AtomicBoolean(false);
	}

	public MultipleKafkaConsumer(String topic, String groudId, Integer consumerCount, KafkaConsumerEventListener listener) {
		this.topic = topic;
		this.groupId = groudId;
		this.consumerCount = consumerCount;
		this.listener = listener;
		this.consumerList = new ConcurrentHashMap<Integer, ConsumerThread>();
		this.isRunning = new AtomicBoolean(false);
	}

	public String getTopic() {
		return this.topic;
	}

	public String getGroupId() {
		return this.groupId;
	}

	@Override
	public void start(Properties properties) {
		if (properties != null && consumerCount > 0 && this.isRunning.compareAndSet(false, true)) {
			properties.setProperty("group.id", this.groupId);
			for (int i = 1; i <= consumerCount; i++) {
				ConsumerThread thread = new ConsumerThread(i, properties);
				thread.start();
				consumerList.put(i, thread);
			}
		}
	}

	@Override
	public void close() {
		this.isRunning.set(false);
	}

	private class ConsumerThread extends Thread {

		final Integer index;
		final Properties properties;

		public ConsumerThread(Integer index, Properties properties) {
			this.index = index;
			this.properties = properties;
		}

		public void run() {
			KafkaConsumer<String, String> kafkaConsumer = new KafkaConsumer<String, String>(properties);
			kafkaConsumer.subscribe(Collections.singletonList(topic));
			while (isRunning.get()) {
				try {
					ConsumerRecords<String, String> records = kafkaConsumer.poll(Duration.ofSeconds(10));
					List<ConsumerEvent> events = new ArrayList<ConsumerEvent>();
					for (ConsumerRecord<String, String> record : records) {
						ConsumerEvent event = new ConsumerEvent();
						event.setKey(record.key());
						event.setTopic(record.topic());
						event.setValue(record.value());
						event.setPartition(record.partition());
						events.add(event);
					}
					
					if (listener != null) {
						try {
							listener.receive(events);
						} catch (Exception e) {
							e.printStackTrace();
						}
					}
					
					kafkaConsumer.commitAsync(); // 手动异步提交( 重点 )
				} catch (Exception e) {
					e.printStackTrace();
				}
			}
			// 关闭KafkaConsumer
			if (kafkaConsumer != null) {
				kafkaConsumer.close();
			}
			consumerList.remove(index);
		}
	}

}
